package org.atlantis.datasink.server;

import java.util.List;
import java.util.concurrent.TimeUnit;

import org.atlantis.datasink.codec.DeviceIdMapping;
import org.atlantis.datasink.codec.ICodec;
import org.atlantis.datasink.codec.ICodecContext;
import org.atlantis.datasink.codec.IDecoder;
import org.atlantis.datasink.codec.IEncoder;
import org.atlantis.datasink.utils.SubstepAwait;
import org.atlantis.datasink.utils.Utils;
import org.jboss.netty.buffer.ChannelBuffer;
import org.jboss.netty.channel.Channel;
import org.jboss.netty.channel.ChannelHandlerContext;
import org.jboss.netty.channel.ChannelPipeline;
import org.jboss.netty.channel.ChannelPipelineFactory;
import org.jboss.netty.channel.Channels;
import org.jboss.netty.handler.codec.frame.FrameDecoder;
import org.jboss.netty.handler.codec.oneone.OneToOneEncoder;
import org.jboss.netty.handler.timeout.ReadTimeoutHandler;

public abstract class AbstractCodecServer implements IServer {

	protected ServerOption option;
	protected DeviceIdMapping mapping;

	@Override
	public void bindOption(ServerOption option) {
		this.option = option;
	}

	@Override
	public void startup(List<ICodecContext> codecs) {
		SubstepAwait await = SubstepAwait.create(codecs.size(), 10 * 1000L);
		for (ICodecContext codecCtx : codecs) {
			final ICodecContext _c = codecCtx;
			await.exec(new Runnable() {
				@Override
				public void run() {
					startByCodec(_c);
				}
			});
		}
		await.waitFinish();
	}

	@Override
	public void setDeviceIdMapping(DeviceIdMapping mapping) {
		this.mapping = mapping;
	}

	protected void startByCodec(ICodecContext codecCtx) {}

	protected ChannelPipelineFactory createPipelineFactory(ICodecContext codecCtx) {
		return new InternalPipelineFactory(codecCtx.getCodec(), codecCtx.getOption().getTimeout());
	}

	private class DecoderHandler extends FrameDecoder {

		private IDecoder decoder;

		DecoderHandler(IDecoder decoder) {
			this.decoder = decoder;
		}

		@Override
		protected Object decode(ChannelHandlerContext ctx, Channel channel, ChannelBuffer buffer) throws Exception {
			return decoder.decode(channel, buffer, mapping);
		}
	}

	private class EncoderHandler extends OneToOneEncoder {

		private IEncoder encoder;

		EncoderHandler(IEncoder encoder) {
			this.encoder = encoder;
		}

		@Override
		protected Object encode(ChannelHandlerContext ctx, Channel channel, Object msg) throws Exception {
			return encoder.encode(channel, msg);
		}
	}

	private class InternalPipelineFactory implements ChannelPipelineFactory {

		private ICodec codec;
		private ReadTimeoutHandler timeoutHandler;

		InternalPipelineFactory(ICodec codec, long timeout) {
			this.codec = codec;
			if (timeout > 0) timeoutHandler = new ReadTimeoutHandler(Utils.shareTimer, timeout, TimeUnit.MILLISECONDS);
		}

		@Override
		public ChannelPipeline getPipeline() throws Exception {
			ChannelPipeline pipeline = Channels.pipeline();
			if (timeoutHandler != null) pipeline.addLast("timeout-handler", timeoutHandler);
			pipeline.addLast(codec.getName() + "-encode-handler", new EncoderHandler(codec.createEncoder()));
			pipeline.addLast("data-io-handler", new DataIoHandler());
			if (codec.createDecoder() != null) pipeline.addLast(codec.getName() + "-decode-handler",
					new DecoderHandler(codec.createDecoder()));
			return pipeline;
		}
	}
}
